#include "destor.h"
#include "jcr.h"
#include "backup.h"
#include "utils/lru_cache.h"
#include "storage/containerstore.h"
#include <openssl/sha.h>
#include <openssl/md5.h>
#include "xdelta/xdelta3.h"
#include "./index/fingerprint_cache.h"
#include "./index/kvstore.h"

typedef unsigned long long int UINT64;

struct SFRecord
{
    fingerprint fp;
    struct superF *superfeature;
    containerid id;
};

struct index_item
{
    fingerprint fp;
    containerid id;
};

struct super_feature_item
{
    uint64_t sf;
    struct index_item *p_item;
};

struct super_feature_item_buffer
{
    uint64_t sf;
    struct chunk *c;
};

struct superFeature
{
    fingerprint fp;
    uint64_t superF;
    containerid id;
};

GHashTable *delta_sf1;
GHashTable *delta_sf2;
GHashTable *delta_sf3;

GHashTable *delta_hash;

GHashTable *delta_sf1_buffer;
GHashTable *delta_sf2_buffer;
GHashTable *delta_sf3_buffer;

GHashTable *delta_hash_buffer;

struct lruCache *base_cache;

extern struct metaEntry
{
    int32_t off;
    int32_t len;
    fingerprint fp;
    /*
     * the flag indicates whether it is a delta
     * 0 -> base chunk
     * 1 -> delta chunk
     * */
    char flag;
    uint64_t sf1;
    uint64_t sf2;
    uint64_t sf3;
    struct metaEntry *pre;
    struct metaEntry *next;
    // containerid baseid;
    // int32_t delta_size;
    // int32_t base_size;
};

static pthread_mutex_t mutex;
static pthread_mutex_t bmutex;

static pthread_t matching_t;

void init_delta_index()
{
    base_cache = new_lru_cache(destor.base_container_num, free_container,
                               lookup_fingerprint_in_container);

    // delta_index = g_hash_table_new_full(g_int64_hash, g_fingerprint_equal, NULL, free);

    delta_sf1 = g_hash_table_new_full(g_int64_hash, g_superfeature_equal, free, NULL);
    delta_sf2 = g_hash_table_new_full(g_int64_hash, g_superfeature_equal, free, NULL);
    delta_sf3 = g_hash_table_new_full(g_int64_hash, g_superfeature_equal, free, NULL);

    delta_hash = g_hash_table_new_full(g_int64_hash, g_fingerprint_equal, NULL, free);

    delta_sf1_buffer = g_hash_table_new_full(g_int64_hash, g_superfeature_equal, free, NULL);
    delta_sf2_buffer = g_hash_table_new_full(g_int64_hash, g_superfeature_equal, free, NULL);
    delta_sf3_buffer = g_hash_table_new_full(g_int64_hash, g_superfeature_equal, free, NULL);

    delta_hash_buffer = g_hash_table_new_full(g_int64_hash, g_fingerprint_equal, NULL, free_chunk);

    sds indexpath = sdsdup(destor.working_directory);
    indexpath = sdscat(indexpath, "index/delta.index");

    sds sf1_path = sdsdup(destor.working_directory);
    sf1_path = sdscat(sf1_path, "index/SF1");

    FILE *fp_sf1;
    if ((fp_sf1 = fopen(sf1_path, "r")))
    {
        /* The number of features */
        UINT64 key_num;
        fread(&key_num, sizeof(UINT64), 1, fp_sf1);
        for (; key_num > 0; key_num--)
        {
            /* Read a feature */
            struct superFeature *p = (struct superFeature *)
                malloc(sizeof(struct superFeature));
            fread(p->fp, sizeof(fingerprint), 1, fp_sf1);
            fread(&p->superF, sizeof(uint64_t), 1, fp_sf1);
            fread(&p->id, sizeof(containerid), 1, fp_sf1);

            struct super_feature_item *item = (struct super_feature_item *)
                malloc(sizeof(struct super_feature_item));

            item->sf = p->superF;
            struct index_item *p_item = g_hash_table_lookup(delta_hash, &p->fp);
            if (!p_item)
            {
                p_item = (struct index_item *)malloc(sizeof(struct index_item));
                memcpy(p_item->fp, p->fp, sizeof(fingerprint));
                p_item->id = p->id;

                g_hash_table_insert(delta_hash, &p_item->fp, p_item);
            }

            item->p_item = p_item;

            g_hash_table_insert(delta_sf1, &item->sf, item);

            free(p);
        }
        fclose(fp_sf1);
    }

    sds sf2_path = sdsdup(destor.working_directory);
    sf2_path = sdscat(sf2_path, "index/SF2");

    FILE *fp_sf2;
    if ((fp_sf2 = fopen(sf2_path, "r")))
    {
        /* The number of features */
        UINT64 key_num;
        fread(&key_num, sizeof(UINT64), 1, fp_sf2);
        for (; key_num > 0; key_num--)
        {
            /* Read a feature */
            struct superFeature *p = (struct superFeature *)
                malloc(sizeof(struct superFeature));
            fread(p->fp, sizeof(fingerprint), 1, fp_sf2);
            fread(&p->superF, sizeof(uint64_t), 1, fp_sf2);
            fread(&p->id, sizeof(containerid), 1, fp_sf2);

            struct super_feature_item *item = (struct super_feature_item *)
                malloc(sizeof(struct super_feature_item));

            item->sf = p->superF;
            struct index_item *p_item = g_hash_table_lookup(delta_hash, &p->fp);
            if (!p_item)
            {
                p_item = (struct index_item *)malloc(sizeof(struct index_item));
                memcpy(p_item->fp, p->fp, sizeof(fingerprint));
                p_item->id = p->id;

                g_hash_table_insert(delta_hash, &p_item->fp, p_item);
            }

            item->p_item = p_item;

            g_hash_table_insert(delta_sf2, &item->sf, item);

            free(p);
        }
        fclose(fp_sf2);
    }

    sds sf3_path = sdsdup(destor.working_directory);
    sf3_path = sdscat(sf3_path, "index/SF3");

    FILE *fp_sf3;
    if ((fp_sf3 = fopen(sf3_path, "r")))
    {
        /* The number of features */
        UINT64 key_num;
        fread(&key_num, sizeof(UINT64), 1, fp_sf3);
        for (; key_num > 0; key_num--)
        {
            /* Read a feature */
            struct superFeature *p = (struct superFeature *)
                malloc(sizeof(struct superFeature));
            fread(p->fp, sizeof(fingerprint), 1, fp_sf3);
            fread(&p->superF, sizeof(uint64_t), 1, fp_sf3);
            fread(&p->id, sizeof(containerid), 1, fp_sf3);

            struct super_feature_item *item = (struct super_feature_item *)
                malloc(sizeof(struct super_feature_item));

            item->sf = p->superF;
            struct index_item *p_item = g_hash_table_lookup(delta_hash, &p->fp);
            if (!p_item)
            {
                p_item = (struct index_item *)malloc(sizeof(struct index_item));
                memcpy(p_item->fp, p->fp, sizeof(fingerprint));
                p_item->id = p->id;

                g_hash_table_insert(delta_hash, &p_item->fp, p_item);
            }

            item->p_item = p_item;

            g_hash_table_insert(delta_sf3, &item->sf, item);

            free(p);
        }
        fclose(fp_sf3);
    }

    sdsfree(indexpath);
    sdsfree(sf1_path);
    sdsfree(sf2_path);
    sdsfree(sf3_path);

    if (pthread_mutex_init(&mutex, 0))
    {
        puts("Failed to init delta mutex!");
        exit(1);
    }

    if (pthread_mutex_init(&bmutex, 0))
    {
        puts("Failed to init delta mutex!");
        exit(1);
    }
}

void super_feature_index_update(GSequence *chunks)
{
    // NOTICE("Filter phase: update %d features", g_sequence_get_length(chunks));
    GSequenceIter *iter = g_sequence_get_begin_iter(chunks);
    GSequenceIter *end = g_sequence_get_end_iter(chunks);
    for (; iter != end; iter = g_sequence_iter_next(iter))
    {
        struct chunk *ck = g_sequence_get(iter);

        if (ck->deltaCompressed)
            continue;

        struct index_item *p_item = g_hash_table_lookup(delta_hash, &ck->fp);
        if (!p_item)
        {
            p_item = (struct index_item *)malloc(sizeof(struct index_item));
            memcpy(p_item->fp, ck->fp, sizeof(fingerprint));
            p_item->id = ck->id;

            g_hash_table_insert(delta_hash, &p_item->fp, p_item);
        }

        struct super_feature_item *item_1 = (struct super_feature_item *)
            malloc(sizeof(struct super_feature_item));
        item_1->sf = ck->superfeature->sf1;
        item_1->p_item = p_item;

        struct super_feature_item *item_2 = (struct super_feature_item *)
            malloc(sizeof(struct super_feature_item));
        item_2->sf = ck->superfeature->sf2;
        item_2->p_item = p_item;

        struct super_feature_item *item_3 = (struct super_feature_item *)
            malloc(sizeof(struct super_feature_item));
        item_3->sf = ck->superfeature->sf3;
        item_3->p_item = p_item;

        pthread_mutex_lock(&mutex);
        g_hash_table_replace(delta_sf1, &item_1->sf, item_1);
        g_hash_table_replace(delta_sf2, &item_2->sf, item_2);
        g_hash_table_replace(delta_sf3, &item_3->sf, item_3);
        pthread_mutex_unlock(&mutex);

        // filter_buffer_unique(ck);
    }
}

/*
 *  Remove the chunks in delta_hash_buffer when they are written in disk
 */
void filter_buffer_unique(struct chunk *ck)
{
    pthread_mutex_lock(&bmutex);
    struct chunk *bc = g_hash_table_lookup(delta_hash_buffer, &ck->fp);
    if (bc)
    {
        g_hash_table_remove(delta_hash_buffer, &ck->fp);
    }

    struct super_feature_item_buffer *bitem_1 = g_hash_table_lookup(delta_sf1_buffer, &ck->superfeature->sf1);
    struct super_feature_item_buffer *bitem_2 = g_hash_table_lookup(delta_sf2_buffer, &ck->superfeature->sf2);
    struct super_feature_item_buffer *bitem_3 = g_hash_table_lookup(delta_sf3_buffer, &ck->superfeature->sf3);

    if (bitem_1)
    {
        g_hash_table_remove(delta_sf1_buffer, &ck->superfeature->sf1);
    }
    if (bitem_2)
    {
        g_hash_table_remove(delta_sf2_buffer, &ck->superfeature->sf2);
    }
    if (bitem_3)
    {
        g_hash_table_remove(delta_sf3_buffer, &ck->superfeature->sf3);
    }
    pthread_mutex_unlock(&bmutex);
}

void close_delta_index()
{

    jcr.total_sf_num = g_hash_table_size(delta_hash);

    sds indexpath = sdsdup(destor.working_directory);
    indexpath = sdscat(indexpath, "index/delta.index");

    sds sf1_path = sdsdup(destor.working_directory);
    sf1_path = sdscat(sf1_path, "index/SF1");

    sds sf2_path = sdsdup(destor.working_directory);
    sf2_path = sdscat(sf2_path, "index/SF2");

    sds sf3_path = sdsdup(destor.working_directory);
    sf3_path = sdscat(sf3_path, "index/SF3");

    FILE *fp_sf1;
    if ((fp_sf1 = fopen(sf1_path, "w")) == NULL)
    {
        perror("Can not open delta.index for write because:");
        exit(1);
    }

    FILE *fp_sf2;
    if ((fp_sf2 = fopen(sf2_path, "w")) == NULL)
    {
        perror("Can not open delta.index for write because:");
        exit(1);
    }

    FILE *fp_sf3;
    if ((fp_sf3 = fopen(sf3_path, "w")) == NULL)
    {
        perror("Can not open delta.index for write because:");
        exit(1);
    }

    GHashTableIter iter;
    gpointer key, value;

    UINT64 key_num = g_hash_table_size(delta_sf1);
    fwrite(&key_num, sizeof(UINT64), 1, fp_sf1);
    g_hash_table_iter_init(&iter, delta_sf1);
    while (g_hash_table_iter_next(&iter, &key, &value))
    {
        struct super_feature_item *p = value;
        fwrite(p->p_item->fp, sizeof(fingerprint), 1, fp_sf1);
        fwrite(&p->sf, sizeof(uint64_t), 1, fp_sf1);
        fwrite(&p->p_item->id, sizeof(containerid), 1, fp_sf1);
    }
    fclose(fp_sf1);

    key_num = g_hash_table_size(delta_sf2);
    fwrite(&key_num, sizeof(UINT64), 1, fp_sf2);
    g_hash_table_iter_init(&iter, delta_sf2);
    while (g_hash_table_iter_next(&iter, &key, &value))
    {
        struct super_feature_item *p = value;
        fwrite(p->p_item->fp, sizeof(fingerprint), 1, fp_sf2);
        fwrite(&p->sf, sizeof(uint64_t), 1, fp_sf2);
        fwrite(&p->p_item->id, sizeof(containerid), 1, fp_sf2);
    }
    fclose(fp_sf2);

    key_num = g_hash_table_size(delta_sf3);
    fwrite(&key_num, sizeof(UINT64), 1, fp_sf3);
    g_hash_table_iter_init(&iter, delta_sf3);
    while (g_hash_table_iter_next(&iter, &key, &value))
    {
        struct super_feature_item *p = value;
        fwrite(p->p_item->fp, sizeof(fingerprint), 1, fp_sf3);
        fwrite(&p->sf, sizeof(uint64_t), 1, fp_sf3);
        fwrite(&p->p_item->id, sizeof(containerid), 1, fp_sf3);
    }
    fclose(fp_sf3);

    sdsfree(indexpath);
    sdsfree(sf1_path);
    sdsfree(sf2_path);
    sdsfree(sf3_path);

    jcr.sketch_index_item_num += g_hash_table_size(delta_sf1);
    jcr.sketch_index_item_num += g_hash_table_size(delta_sf2);
    jcr.sketch_index_item_num += g_hash_table_size(delta_sf3);

    // g_hash_table_destroy(delta_index);
    g_hash_table_destroy(delta_sf1);
    g_hash_table_destroy(delta_sf2);
    g_hash_table_destroy(delta_sf3);

    g_hash_table_destroy(delta_hash);

    g_hash_table_destroy(delta_sf1_buffer);
    g_hash_table_destroy(delta_sf2_buffer);
    g_hash_table_destroy(delta_sf3_buffer);

    g_hash_table_destroy(delta_hash_buffer);
}

void clear_delta_index()
{
    g_hash_table_remove_all(delta_sf1);
    g_hash_table_remove_all(delta_sf2);
    g_hash_table_remove_all(delta_sf3);
}

struct chunk *retrive_basechunk_by_fp_id(fingerprint *fp, containerid id, int enable_retrieve)
{
    struct container *con = NULL;
    struct chunk *c = NULL;
    int32_t res;

    /*
    unsigned char buf[41];
    hash2code(*fp, buf);
    buf[40] = 0;
    if (buf[0] == 'D' && buf[1] == '7' && buf[2] == '8' && buf[3] == '6' && buf[4] == '0' && buf[5] == '7') {
        jcr.zero_chunk_num++;
        printf("fp: %s\n", buf);
    }
    */

    /* first, we find it in the buffer */
    assert(id <= jcr.container_num_now_stored);
    con = retrieve_container_in_write_buffer(id);

    /* then, find it in base cache */
    if (!con)
    {
        res = check_restore_container(base_cache, fp, id);

        if (!res)
        {
            if (!enable_retrieve)
                return c;
            VERBOSE("Restore cache: container %lld is missed", id);
            con = retrieve_container_by_id(id);
            lru_cache_insert(base_cache, con, NULL, NULL);

            jcr.base_cache_miss++;
            jcr.base_container_read++;
        }
        else
        {
            con = lru_cache_lookup(base_cache, fp);
            jcr.base_cache_hit++;
        }
    }
    else
    {
        lru_cache_insert(base_cache, con, NULL, NULL);
        con = lru_cache_lookup(base_cache, fp);
        jcr.base_cache_hit++;
    }

    assert(con);
    c = get_basechunk_in_container(con, fp);

    assert(c);
    return c;
}

struct chunk *retrive_basechunk_incache_by_fp_id(fingerprint *fp, containerid id)
{
    struct container *con = lru_cache_lookup_without_update(base_cache, fp);
    assert(con);

    struct chunk *c = get_basechunk_in_container(con, fp);

    assert(c);
    return c;
}

struct chunk *get_simi_chunk(struct chunk *srcChunk)
{
    struct chunk *baseChunk = NULL;
    struct chunk *bufferChunk = NULL;

    struct super_feature_item *psSuperF = NULL;
    struct super_feature_item *psSuperF1 = NULL;
    struct super_feature_item *psSuperF2 = NULL;
    struct super_feature_item *psSuperF3 = NULL;

    struct super_feature_item_buffer *buffersf1 = NULL;
    struct super_feature_item_buffer *buffersf2 = NULL;
    struct super_feature_item_buffer *buffersf3 = NULL;

    /*
    buffersf1 = g_hash_table_lookup(delta_sf1_buffer, &srcChunk->superfeature->sf1);
    buffersf2 = g_hash_table_lookup(delta_sf2_buffer, &srcChunk->superfeature->sf2);
    buffersf3 = g_hash_table_lookup(delta_sf3_buffer, &srcChunk->superfeature->sf3);
    if (buffersf1)
        bufferChunk = buffersf1->c;
    else if (buffersf2)
        bufferChunk = buffersf2->c;
    else if (buffersf3)
        bufferChunk = buffersf3->c;

    if (bufferChunk)
    {
        baseChunk = new_chunk(bufferChunk->size);
        memcpy(baseChunk->data, bufferChunk->data, bufferChunk->size);
        memcpy(baseChunk->fp, bufferChunk->fp, sizeof(fingerprint));
        baseChunk->id = bufferChunk->id;
    }

    if (baseChunk)
    {
        jcr.sf_matched_number++;
        return baseChunk;
    }
    */

    psSuperF1 = g_hash_table_lookup(delta_sf1, &srcChunk->superfeature->sf1);
    psSuperF2 = g_hash_table_lookup(delta_sf2, &srcChunk->superfeature->sf2);
    psSuperF3 = g_hash_table_lookup(delta_sf3, &srcChunk->superfeature->sf3);

    if (psSuperF2 && psSuperF3 && !memcmp(psSuperF2->p_item->fp, psSuperF3->p_item->fp, destor.index_key_size))
        psSuperF = psSuperF2;
    else if (psSuperF1)
        psSuperF = psSuperF1;
    else if (psSuperF2)
        psSuperF = psSuperF2;
    else if (psSuperF3)
        psSuperF = psSuperF3;
    /*
    if (destor.enable_bestmatch) {
        if (destor.enable_sort_features) {
            if (psSuperF1 && psSuperF2 && !memcmp(psSuperF1->p_item->fp, psSuperF2->p_item->fp, destor.index_key_size))
                psSuperF = psSuperF1;
            else if (psSuperF1 && psSuperF3 && !memcmp(psSuperF1->p_item->fp, psSuperF3->p_item->fp, destor.index_key_size))
                psSuperF = psSuperF1;
            else if (psSuperF2 && psSuperF3 && !memcmp(psSuperF2->p_item->fp, psSuperF3->p_item->fp, destor.index_key_size))
                psSuperF = psSuperF2;
        }
    }

    if ((psSuperF1 && psSuperF2)
        || (psSuperF1 && psSuperF3)
        || (psSuperF2 && psSuperF3)) {
        jcr.multi_matched_chunk_number++;
    }

    if ((psSuperF1 && psSuperF2 && memcmp(psSuperF1->p_item->fp, psSuperF2->p_item->fp, destor.index_key_size))
        || (psSuperF1 && psSuperF3 && memcmp(psSuperF1->p_item->fp, psSuperF3->p_item->fp, destor.index_key_size))
        || (psSuperF2 && psSuperF3 && memcmp(psSuperF2->p_item->fp, psSuperF3->p_item->fp, destor.index_key_size))) {
        jcr.multi_matched_diff_number++;
    }
    */

    if (psSuperF)
    {
        baseChunk = retrive_basechunk_by_fp_id(&psSuperF->p_item->fp, psSuperF->p_item->id, 1);
        jcr.sf_matched_number++;
    }

    return baseChunk;
}

struct chunk *get_simi_chunk_sidc(struct chunk *srcChunk)
{
    struct chunk *baseChunk = NULL;

    struct index_item *item = match_cache_lookup(srcChunk); // index_item
    if (!item)
        return NULL;
    baseChunk = retrive_basechunk_by_fp_id(&item->fp, item->id, 1);
    jcr.sf_matched_number++;

    free(item);
    return baseChunk;
}

struct super_feature_item *find_best_chunk(struct super_feature_item *f1, struct super_feature_item *f2, struct super_feature_item *f3)
{
    int op = simiulated_fingerprint_cache_lookup(f1, f2, f3);
    if (op == 1)
        return f1;
    else if (op == 2)
        return f2;
    else
        return f3;
}

struct chunk *get_simi_chunk_lcs(struct chunk *srcChunk)
{
    struct super_feature_item *psSuperF = NULL;
    struct super_feature_item *psSuperF1 = NULL;
    struct super_feature_item *psSuperF2 = NULL;
    struct super_feature_item *psSuperF3 = NULL;
    struct chunk *baseChunk = NULL;
    struct chunk *baseChunk1 = NULL;
    struct chunk *baseChunk2 = NULL;
    struct chunk *baseChunk3 = NULL;

    psSuperF1 = g_hash_table_lookup(delta_sf1, &srcChunk->superfeature->sf1);
    psSuperF2 = g_hash_table_lookup(delta_sf2, &srcChunk->superfeature->sf2);
    psSuperF3 = g_hash_table_lookup(delta_sf3, &srcChunk->superfeature->sf3);

    if (!psSuperF1 && !psSuperF2 && !psSuperF3)
        return NULL;
    else if (psSuperF1 && !psSuperF2 && !psSuperF3)
        psSuperF = psSuperF1;
    else if (psSuperF2 && !psSuperF1 && !psSuperF3)
        psSuperF = psSuperF2;
    else if (psSuperF3 && !psSuperF1 && !psSuperF2)
        psSuperF = psSuperF3;
    else
        psSuperF = find_best_chunk(psSuperF1, psSuperF2, psSuperF3);
    if (psSuperF)
    {
        baseChunk = retrive_basechunk_by_fp_id(&psSuperF->p_item->fp, psSuperF->p_item->id, 1);
        jcr.sf_matched_number++;
    }

    return baseChunk;
}

/*
 * For the unique chunks that can be used as base chunks in pipeline
 * But the throughput will be affected
 */
void update_buffer_unique_chunk(struct chunk *c)
{
    SET_CHUNK(c, CHUNK_BUFFER_DELTA);

    struct super_feature_item_buffer *item_1 = (struct super_feature_item_buffer *)
        malloc(sizeof(struct super_feature_item_buffer));
    struct super_feature_item_buffer *item_2 = (struct super_feature_item_buffer *)
        malloc(sizeof(struct super_feature_item_buffer));
    struct super_feature_item_buffer *item_3 = (struct super_feature_item_buffer *)
        malloc(sizeof(struct super_feature_item_buffer));

    item_1->sf = c->superfeature->sf1;
    item_2->sf = c->superfeature->sf2;
    item_3->sf = c->superfeature->sf3;

    pthread_mutex_lock(&bmutex);

    struct chunk *bc = g_hash_table_lookup(delta_hash_buffer, &c->fp);
    if (!bc)
    {
        bc = new_chunk(c->size);
        memcpy(bc->fp, c->fp, sizeof(fingerprint));
        bc->id = c->id;
        assert(bc->id == -1);
        memcpy(bc->data, c->data, c->size);
        g_hash_table_insert(delta_hash_buffer, &bc->fp, bc);
    }

    item_1->c = bc;
    item_2->c = bc;
    item_3->c = bc;

    g_hash_table_replace(delta_sf1_buffer, &item_1->sf, item_1);
    g_hash_table_replace(delta_sf2_buffer, &item_2->sf, item_2);
    g_hash_table_replace(delta_sf3_buffer, &item_3->sf, item_3);

    pthread_mutex_unlock(&bmutex);
}

static void *matching_thread(void *arg)
{

    while (1)
    {
        struct chunk *c = sync_queue_pop(resemblance_queue);

        if (c == NULL)
            /* backup job finish */
            break;

        if (CHECK_CHUNK(c, CHUNK_FILE_START) ||
            CHECK_CHUNK(c, CHUNK_FILE_END) ||
            CHECK_CHUNK(c, CHUNK_SEGMENT_START) ||
            CHECK_CHUNK(c, CHUNK_SEGMENT_END))
        {
            sync_queue_push(matching_queue, c);
            continue;
        }
        if (CHECK_CHUNK(c, CHUNK_DUPLICATE))
        {
            sync_queue_push(matching_queue, c);
            continue;
        }

        struct chunk *baseChunk = NULL;
        if (c->size > destor.chunk_min_size)
        {
            TIMER_DECLARE(1);
            TIMER_BEGIN(1);
            baseChunk = get_simi_chunk(c);
            TIMER_END(1, jcr.matching_time);
        }
        int sizediff;
        if (baseChunk)
        {
            if (baseChunk->size > c->size)
                sizediff = (baseChunk->size - c->size) / (1.0 * baseChunk->size);
            else
                sizediff = (c->size - baseChunk->size) / (1.0 * c->size);
            if (destor.super_feature_method == SUPER_FEATURE)
            {
                if (sizediff > 0.05)
                {
                    free(baseChunk);
                    baseChunk = NULL;
                }
            }
        }

        if (baseChunk == NULL)
        {
            c->delta = NULL;
            c->baseChunk = NULL;
            c->type = CHUNK_TYPE_UNIQUE;
            // update_buffer_unique_chunk(c);
        }
        else
        {
            c->baseChunk = baseChunk;
            destor.now_delta_num++;
            c->type = CHUNK_TYPE_DELTA;
        }
        sync_queue_push(matching_queue, c);
    }
    sync_queue_term(matching_queue);
    return NULL;
}

void send_segment_delta(struct segment *s)
{
    struct chunk *ss = new_chunk(0);
    SET_CHUNK(ss, CHUNK_SEGMENT_START);
    sync_queue_push(matching_queue, ss);

    GSequenceIter *end = g_sequence_get_end_iter(s->chunks);
    GSequenceIter *begin = g_sequence_get_begin_iter(s->chunks);
    while (begin != end)
    {
        struct chunk *c = g_sequence_get(begin);
        sync_queue_push(matching_queue, c);
        g_sequence_remove(begin);
        begin = g_sequence_get_begin_iter(s->chunks);
    }

    struct chunk *se = new_chunk(0);
    SET_CHUNK(se, CHUNK_SEGMENT_END);
    sync_queue_push(matching_queue, se);
    s->chunk_num = 0;
}

int hit_num, lookup_num, miss_num;
int adjacent_num;

void get_adjacent_chunk(struct segment *s, struct chunk *lastChunk, struct chunk *nowChunk,
                        GSequenceIter *last_iter, GSequenceIter *now_iter, int last_index, int now_index)
{
    if (destor.index_category[1] == INDEX_CATEGORY_LOGICAL_LOCALITY)
    {
        int preMatchLen = 0, nextMatchLen = 0, i;
        struct chunkPointer *last_cp = NULL, *now_cp = NULL;

        if (lastChunk && lastChunk->baseChunk)
            lastChunk = lastChunk->baseChunk;

        if (nowChunk->baseChunk)
            nowChunk = nowChunk->baseChunk;

        if (lastChunk)
            last_cp = chunkpointer_cache_lookup(lastChunk); // The sr has been prefetched in last call

        now_cp = chunkpointer_cache_lookup(nowChunk);
        lookup_num++;
        if (!now_cp)
        {
            miss_num++;
            int64_t *ids = kvstore_lookup((char *)&nowChunk->fp);
            if (ids)
            {
                chunkpointer_cache_prefetch(ids[0]);
                now_cp = chunkpointer_cache_lookup(nowChunk);
            }
        }

        if (now_cp == NULL && last_cp == NULL)
            return;

        if (now_cp == NULL) // Now chunk can't be found
        {
            preMatchLen = 0;
            nextMatchLen = min(destor.adjchunk_match_length, now_index - last_index - 1);
        }
        else if (last_cp == NULL) // First similar chunk or last chunk can't be found
        {
            nextMatchLen = 0;
            if (last_index < 0)
                preMatchLen = min(destor.adjchunk_match_length, now_index);
            else
                preMatchLen = min(destor.adjchunk_match_length, now_index - last_index - 1);
        }
        else
        {
            preMatchLen = min(destor.adjchunk_match_length, (now_index - last_index) / 2);
            nextMatchLen = min(destor.adjchunk_match_length, now_index - last_index - 1); // Greedy match
        }

        int lv_adjacent = 1;
        GSequenceIter *cur_iter = now_iter;

        for (i = 1; i <= preMatchLen; ++i)
        {
            struct chunk *preChunk = NULL;
            struct chunkPointer *lookupCp = NULL;
            assert(now_index - i >= 0);
            cur_iter = g_sequence_iter_prev(cur_iter);
            preChunk = g_sequence_get(cur_iter);

            if (CHECK_CHUNK(preChunk, CHUNK_FILE_START) ||
                CHECK_CHUNK(preChunk, CHUNK_FILE_END) ||
                CHECK_CHUNK(preChunk, CHUNK_ADJ_UNIQUE))
                break;

            if (preChunk->type == CHUNK_TYPE_UNIQUE)
            {
                lookupCp = now_cp;
                int j = 0;
                while (lookupCp && j++ < i)
                    lookupCp = lookupCp->pre;

                if (!lookupCp)
                    break;
                if (lookupCp->flag == CP_TYPE_CHUNK)
                {
                    assert(lookupCp->id <= jcr.container_num_now_stored + 1);
                    if (lookupCp->id <= jcr.container_num_now_stored)
                    {
                        preChunk->baseChunk = retrive_basechunk_by_fp_id(&lookupCp->fp, lookupCp->id, 1);
                        SET_CHUNK(preChunk, CHUNK_ADJ_UNIQUE_LEFT);
                        SET_CHUNK(preChunk, CHUNK_ADJ_UNIQUE);
                        preChunk->type = CHUNK_TYPE_DELTA;
                        preChunk->level = lv_adjacent++;
                        adjacent_num++;
                    }
                }
            }
        }

        lv_adjacent = 1;
        cur_iter = last_iter;
        int len = g_sequence_get_length(s->chunks);
        for (i = 1; i <= nextMatchLen && last_index + i < len; ++i)
        {
            struct chunk *nextChunk = NULL;
            struct chunkPointer *lookupCp = NULL;
            cur_iter = g_sequence_iter_next(cur_iter);
            nextChunk = g_sequence_get(cur_iter);

            if (CHECK_CHUNK(nextChunk, CHUNK_FILE_START) ||
                CHECK_CHUNK(nextChunk, CHUNK_FILE_END) ||
                CHECK_CHUNK(nextChunk, CHUNK_ADJ_UNIQUE))
                break;

            if (nextChunk->type == CHUNK_TYPE_UNIQUE)
            {
                lookupCp = last_cp;
                int j = 0;
                while (lookupCp && j++ < i)
                    lookupCp = lookupCp->next;

                if (!lookupCp)
                    break;
                if (lookupCp->flag == CP_TYPE_CHUNK)
                {
                    assert(lookupCp->id <= jcr.container_num_now_stored + 1);
                    if (lookupCp->id <= jcr.container_num_now_stored)
                    {
                        nextChunk->baseChunk = retrive_basechunk_by_fp_id(&lookupCp->fp, lookupCp->id, 1);
                        SET_CHUNK(nextChunk, CHUNK_ADJ_UNIQUE_RIGHT);
                        SET_CHUNK(nextChunk, CHUNK_ADJ_UNIQUE);
                        nextChunk->type = CHUNK_TYPE_DELTA;
                        nextChunk->level = lv_adjacent++;
                        adjacent_num++;
                    }
                }
            }
        }
    }
    else
    {
        if (lastChunk && lastChunk->baseChunk)
            lastChunk = lastChunk->baseChunk;

        if (nowChunk->baseChunk)
            nowChunk = nowChunk->baseChunk;

        int MatchNum = match_adjacent(s, lastChunk, nowChunk, last_iter, now_iter, last_index, now_index);
        int i = 0;
        GSequenceIter *cur_iter = now_iter;
        while (i < MatchNum)
        {
            cur_iter = g_sequence_iter_prev(cur_iter);
            assert(cur_iter);

            struct chunk *preChunk = g_sequence_get(cur_iter);
            if (CHECK_CHUNK(preChunk, CHUNK_ADJ_UNIQUE))
            {
                preChunk->baseChunk = retrive_basechunk_by_fp_id(&preChunk->basefp, preChunk->baseid, 1);
                i++;
            }
        }
    }
}

void get_last_adjacent_chunk(GSequenceIter *last_iter, struct segment *s, int index)
{
    if (destor.index_category[1] == INDEX_CATEGORY_LOGICAL_LOCALITY)
    {
        int len = g_sequence_get_length(s->chunks), i;
        struct chunk *c = g_sequence_get(last_iter);
        if (c->baseChunk)
            c = c->baseChunk;
        struct chunkPointer *now_cp = chunkpointer_cache_lookup(c);
        int nextMatchLen = 0;
        if (now_cp != NULL) // Now chunk can't be found
        {
            nextMatchLen = min(destor.adjchunk_match_length, len - index - 1);
        }

        int lv_adjacent = 1;
        GSequenceIter *cur_iter = last_iter;
        for (i = 1; i <= nextMatchLen; ++i)
        {
            struct chunk *nextChunk = NULL;
            struct chunkPointer *lookupCp = NULL;
            assert(index + i < len);
            cur_iter = g_sequence_iter_next(cur_iter);
            nextChunk = g_sequence_get(cur_iter);

            if (nextChunk && nextChunk->type == CHUNK_TYPE_UNIQUE)
            {
                lookupCp = now_cp;
                int j = 0;
                while (lookupCp && j++ < i)
                    lookupCp = lookupCp->next;

                if (lookupCp && lookupCp->flag == CP_TYPE_CHUNK)
                {
                    assert(lookupCp->id <= jcr.container_num_now_stored + 1);
                    if (lookupCp->id <= jcr.container_num_now_stored)
                    {
                        nextChunk->baseChunk = retrive_basechunk_by_fp_id(&lookupCp->fp, lookupCp->id, 1);
                        SET_CHUNK(nextChunk, CHUNK_ADJ_UNIQUE_RIGHT);
                        SET_CHUNK(nextChunk, CHUNK_ADJ_UNIQUE);
                        nextChunk->type = CHUNK_TYPE_DELTA;
                        nextChunk->level = lv_adjacent++;
                        adjacent_num++;
                    }
                }
            }
        }
    }
    else
    {
        struct chunk *c = g_sequence_get(last_iter);
        if (c->baseChunk)
            c = c->baseChunk;
        int MatchNum = match_last_adjacent(s, c, last_iter, index);
        int i = 0;
        GSequenceIter *cur_iter = last_iter;
        while (i < MatchNum)
        {
            cur_iter = g_sequence_iter_next(cur_iter);
            assert(cur_iter);

            struct chunk *nextChunk = g_sequence_get(cur_iter);
            if (CHECK_CHUNK(nextChunk, CHUNK_ADJ_UNIQUE))
            {
                nextChunk->baseChunk = retrive_basechunk_by_fp_id(&nextChunk->basefp, nextChunk->baseid, 1);
                i++;
            }
        }
    }
}

/*
 * The unit is segment which is suitable to match the adjacent chunks.
 */
static void *matching_thread_segment(void *arg)
{
    while (1)
    {
        struct chunk *c = sync_queue_pop(resemblance_queue);

        if (c == NULL)
            /* backup job finish */
            break;

        // sync_queue_push(matching_queue, c);
        // continue;

        struct segment *s = new_segment();
        assert(CHECK_CHUNK(c, CHUNK_SEGMENT_START));
        free_chunk(c);

        c = sync_queue_pop(resemblance_queue);
        while (!(CHECK_CHUNK(c, CHUNK_SEGMENT_END)))
        {
            g_sequence_append(s->chunks, c);

            if (!CHECK_CHUNK(c, CHUNK_FILE_START) && !CHECK_CHUNK(c, CHUNK_FILE_END))
                s->chunk_num++;

            c = sync_queue_pop(resemblance_queue);
        }

        free_chunk(c);

        GSequenceIter *last_iter = NULL;

        int i = 0, index = -1;

        GSequenceIter *iter = g_sequence_get_begin_iter(s->chunks);
        GSequenceIter *end = g_sequence_get_end_iter(s->chunks);
        for (; iter != end; iter = g_sequence_iter_next(iter), i++)
        {
            struct chunk *c = g_sequence_get(iter);
            struct chunk *baseChunk = NULL;
            struct chunk *lastChunk = NULL;

            if (CHECK_CHUNK(c, CHUNK_FILE_START) || CHECK_CHUNK(c, CHUNK_FILE_END))
            {
                continue;
            }

            if (CHECK_CHUNK(c, CHUNK_DUPLICATE))
            {
                TIMER_DECLARE(1);
                TIMER_BEGIN(1);
                if (destor.enable_match_adjacent && index + 1 != i)
                {
                    lastChunk = last_iter != NULL ? g_sequence_get(last_iter) : NULL;
                    get_adjacent_chunk(s, lastChunk, c, last_iter, iter, index, i);
                }

                TIMER_END(1, jcr.matching_time);
                last_iter = iter;
                index = i;
                continue;
            }

            if (c->size > destor.chunk_min_size)
            {
                TIMER_DECLARE(1);
                TIMER_BEGIN(1);
                if (destor.enable_sidc)
                    baseChunk = get_simi_chunk_sidc(c);
                else
                    baseChunk = get_simi_chunk(c);
                TIMER_END(1, jcr.matching_time);
            }
            if (baseChunk == NULL)
            {
                c->delta = NULL;
                c->baseChunk = NULL;
                c->type = CHUNK_TYPE_UNIQUE;
            }
            else
            {
                c->baseChunk = baseChunk;
                destor.now_delta_num++;
                c->type = CHUNK_TYPE_DELTA;
                c->level = 1;

                TIMER_DECLARE(1);
                TIMER_BEGIN(1);
                if (destor.enable_match_adjacent && index + 1 != i)
                {
                    lastChunk = last_iter != NULL ? g_sequence_get(last_iter) : NULL;
                    get_adjacent_chunk(s, lastChunk, c, last_iter, iter, index, i);
                }
                TIMER_END(1, jcr.matching_time);
                last_iter = iter;
                index = i;
            }
        }

        TIMER_DECLARE(1);
        TIMER_BEGIN(1);
        if (destor.enable_match_adjacent && last_iter != NULL)
            get_last_adjacent_chunk(last_iter, s, index);
        TIMER_END(1, jcr.matching_time);

        send_segment_delta(s);
        free_segment(s);
        s = NULL;
    }
    sync_queue_term(matching_queue);
    return NULL;
}

void start_matching_phase()
{
    init_delta_index();
    init_chunkpointer_cache();
    matching_queue = sync_queue_new(1000);
    pthread_create(&matching_t, NULL, matching_thread_segment, NULL);
}

void stop_matching_phase()
{
    hit_num = lookup_num - miss_num;
    printf("lookup_num rate %lf, hit_num is %d miss_num is %d\n", hit_num * 1.0 / lookup_num, hit_num, miss_num);
    printf("All adjacent num %d\n", adjacent_num);
    pthread_join(matching_t, NULL);
    free_lru_cache(base_cache);
}
